package com.walkud.rxjava.combining;

import com.walkud.rxjava.utils.PrintLogSubscriber;
import com.walkud.rxjava.utils.RxSubscribe;
import com.walkud.rxjava.utils.SleepUtil;
import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;

import java.util.concurrent.TimeUnit;

/**
 * 组合操作示例
 * <p>
 * And/Then/When — combine sets of items emitted by two or more Observables by means of Pattern and Plan intermediaries
 * CombineLatest — when an item is emitted by either of two Observables, combine the latest item emitted by each Observable via a specified function and emit items based on the results of this function
 * Join — combine items emitted by two Observables whenever an item from one Observable is emitted during a time window defined according to an item emitted by the other Observable
 * Merge — combine multiple Observables into one by merging their emissions
 * StartWith — emit a specified sequence of items before beginning to emit the items from the source Observable
 * Switch — convert an Observable that emits Observables into a single Observable that emits the items emitted by the most-recently-emitted of those Observables
 * Zip — combine the emissions of multiple Observables together via a specified function and emit single items for each combination based on the results of this function
 */
public class CombiningSample {

    public static void main(String[] args) {
//        And/Then/When()//没找到对应的操作符

//        combineLatest();
//        join();
//        merge();
//        startWith();
//        Switch();
//        zip();
    }

    /**
     * 通过一个函数将多个Observables的发射物结合到一起，基于这个函数的结果为每个结合体发射单个数据项。
     */
    private static void zip() {
        //产生1，2，3
        Observable<Long> observable1 = Observable.interval(1, TimeUnit.SECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) throws Exception {
                        return aLong + 1;
                    }
                }).take(3);
        //产生4，5，6
        Observable<Long> observable2 = Observable.interval(1, TimeUnit.SECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) throws Exception {
                        return aLong + 4;
                    }
                }).take(3);

        Observable.zip(observable1, observable2, new BiFunction<Long, Long, Long>() {
            @Override
            public Long apply(Long aLong, Long aLong2) throws Exception {
                return aLong + aLong2;
            }
        }).blockingSubscribe(PrintLogSubscriber.create());
    }

    /**
     * 把一组Observable转换成一个Observable，转换规则为：对于这组Observable中的每一个Observable所产生的结果，如果在同一个时间内存在两个或多个Observable提交的结果，只取最后一个Observable提交的结果给订阅者。
     */
    private static void Switch() {
        Observable<Observable<Long>> observable = Observable.interval(500, TimeUnit.MILLISECONDS).map(new Function<Long, Observable<Long>>() {
            @Override
            public Observable<Long> apply(Long aLong) {
                //每隔200毫秒产生一组数据（0,10,20,30,40)
                return Observable.interval(200, TimeUnit.MILLISECONDS).map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) {
                        return aLong * 10;
                    }
                }).take(3);
            }
        }).take(2);

//        observable.flatMap(new Function<Observable<Long>, ObservableSource<?>>() {
//            @Override
//            public ObservableSource<Long> apply(Observable<Long> longObservable) throws Exception {
//                return longObservable;
//            }
//        }).blockingSubscribe(PrintLogSubscriber.create());
        Observable.switchOnNext(observable).blockingSubscribe(PrintLogSubscriber.create());
    }


    /**
     * 在发射数据序列的开头插入一些指定的数据项
     */
    private static void startWith() {
        Observable<Integer> observable1 = Observable.interval(1, TimeUnit.SECONDS).map(new Function<Long, Integer>() {
            @Override
            public Integer apply(Long aLong) throws Exception {
                return aLong.intValue();
            }
        }).take(3);
//        Observable<Integer> observable1 = Observable.just(1);
        Observable<Integer> observable2 = Observable.just(2, 3, 4, 5);

        observable2.startWith(observable1).blockingSubscribe(PrintLogSubscriber.create());
    }

    /**
     * 将多个Observable合并为一个,按时间顺序合并发射
     */
    private static void merge() {
//        Observable<Integer> observable1 = Observable.just(1, 2, 3);
//        Observable<Integer> observable2 = Observable.just(4, 5, 6);

        //产生1，2，3
        Observable<Long> observable1 = Observable.interval(1, TimeUnit.SECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) throws Exception {
                        return aLong + 1;
                    }
                }).take(3);
        //产生4，5，6
        Observable<Long> observable2 = Observable.interval(1, TimeUnit.SECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) throws Exception {
                        return aLong + 4;
                    }
                }).take(3);

        observable1.mergeWith(observable2).blockingSubscribe(PrintLogSubscriber.create());

    }

    /**
     * 任何时候，只要在另一个Observable发射的数据定义的时间窗口内，这个Observable发射了一条数据，就结合两个Observable发射的数据。
     */
    private static void join() {
        //创建被观察者1，产生0，5，10
        Observable<Long> observable1 = Observable.interval(1500, TimeUnit.MILLISECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) throws Exception {
                        return aLong * 5;
                    }
                }).take(3);
        //创建被观察者2，产生0，10，20
        Observable<Long> observable2 = Observable.interval(1500, TimeUnit.MILLISECONDS)
                .map(new Function<Long, Long>() {
                    @Override
                    public Long apply(Long aLong) throws Exception {
                        return aLong * 10;
                    }
                }).take(3);


        observable1.join(observable2, new Function<Long, ObservableSource<Long>>() {
            @Override
            public ObservableSource<Long> apply(Long aLong) throws Exception {
                //Funcation1 接收观察者1发射的数据，对发射数据进行有效期控制
                System.out.println("aLong1:" + aLong);
                return Observable.just(aLong).delay(500, TimeUnit.MILLISECONDS);
            }
        }, new Function<Long, ObservableSource<Long>>() {
            @Override
            public ObservableSource<Long> apply(Long aLong) throws Exception {
                //Funcation2 接收观察者2发射的数据，对发射数据进行有效期控制
                System.out.println("aLong2:" + aLong);
                return Observable.just(aLong).delay(500, TimeUnit.MILLISECONDS);
            }
        }, new BiFunction<Long, Long, Long>() {
            @Override
            public Long apply(Long aLong1, Long aLong2) throws Exception {
                //组合观察者1和观察下和2控制后的数据
                System.out.println("aLong1:" + aLong1 + ",  aLong2:" + aLong2);
                return aLong1 + aLong2;
            }
        }).blockingSubscribe(PrintLogSubscriber.create());
    }

    /**
     * 当两个Observables中的任何一个发射了数据时，使用一个函数结合每个Observable发射的最近数据项，并且基于这个函数的结果发射数据。
     * 使用场景:网上多数使用combineLatest来做表单验证，多个输入框是否同时为空判断
     */
    private static void combineLatest() {
        Observable.combineLatest(Observable.just(1, 2, 3), Observable.just(4, 5, 6), new BiFunction<Integer, Integer, String>() {
            @Override
            public String apply(Integer integer, Integer integer2) throws Exception {
                return integer + "+" + integer2 + "=" + (integer + integer2);
            }
        }).subscribe(PrintLogSubscriber.create());
    }

}
